package com.zuikaku.demo;

import com.zuikaku.pojo.Order;
import com.zuikaku.source.OrderSource;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class _4TumblingWindowDemoWithProcessTime {

    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment environment =StreamExecutionEnvironment.createLocalEnvironment();
        //2.指定模式Auto并设置并行度
        environment.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        environment.setParallelism(1);
        //3.指定source
        DataStream<Order> orderDs = environment.addSource(new OrderSource());
        //4.以itemName进行分组
        KeyedStream<Order, String> keyedStream = orderDs.keyBy(new KeySelector<Order, String>() {
            @Override
            public String getKey(Order order) throws Exception {
                return order.getItemName();
            }
        });
        //5.开窗使用时间滚动窗口，每5s统计过去5s的，并在全窗口函数中求和
        DataStream<Order> applied = keyedStream.window(TumblingProcessingTimeWindows.of(Time.of(5, TimeUnit.SECONDS))).process(new ProcessWindowFunction<Order, Order, String, TimeWindow>() {
            @Override
            public void process(String key, ProcessWindowFunction<Order, Order, String, TimeWindow>.Context context, Iterable<Order> input, Collector<Order> collector) throws Exception {

                //其中key是分组的那个字段值，iterable为这组窗内数据的迭代器，collector用于收集对外输出
                //5.1通过迭代器拿到这组窗内数据
                List<Order> orderInWindow = IteratorUtils.toList(input.iterator());
                //5.2做求和
                BigDecimal sum = new BigDecimal(0);
                for (Order o:orderInWindow) {
                    sum =sum.add(o.getPrice());
                }
                //5.3准备out对象
                Order out = new Order();
                out.setPrice(sum);
                out.setItemName(key);
                out.setCreateDate(new Date(context.window().getEnd()));//window结束时间便于观察
                collector.collect(out);
            }
        });

        applied.print("全窗口：");

        try {
            environment.execute("apply window");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
